package com.atguigu.bigdata.edu.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;

import com.atguigu.bigdata.edu.realtime.app.BaseAppV1;
import com.atguigu.bigdata.edu.realtime.bean.TradeCourseOrderBean;
import com.atguigu.bigdata.edu.realtime.common.Constant;
import com.atguigu.bigdata.edu.realtime.util.AtguiguUtil;
import com.atguigu.bigdata.edu.realtime.util.DimUtil;
import com.atguigu.bigdata.edu.realtime.util.JdbcUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;
import java.sql.Connection;
import java.time.Duration;

/**
 * @Author lzc
 * @Date 2022/10/18 10:36
 */
public class Dws_09_DwsTradeCourseOrderWindow extends BaseAppV1 {
    public static void main(String[] args) {
        new Dws_09_DwsTradeCourseOrderWindow().init(
                4009,
                2,
                "Dws_09_DwsTradeCourseOrderWindow",
                Constant.TOPIC_DWD_TRADE_ORDER_DETAIL
        );

    }

    @Override
    protected void handle(StreamExecutionEnvironment env,
                          DataStreamSource<String> stream) {
        // 1. 按照 订单详情 id 去重
        stream.print("stream");
        SingleOutputStreamOperator<JSONObject> distinctedStream = distinct(stream);
        distinctedStream.print("distinctedStream");
        // 2. 把数据封装到 pojo 中
        SingleOutputStreamOperator<TradeCourseOrderBean> beanStream = parseToPOJO(distinctedStream);
        beanStream.print("beanStream");
        // 3. 开窗聚合
        SingleOutputStreamOperator<TradeCourseOrderBean> aggregatedStreamWithoutDims = windowAndAgg(beanStream);
        aggregatedStreamWithoutDims.print("aggregatedStreamWithoutDims");

        // 4. 补充维度信息
        addDims(aggregatedStreamWithoutDims);


        // 5. 写出到clickhouse 中
    }

    private void addDims(SingleOutputStreamOperator<TradeCourseOrderBean> stream) {
        // 补充维度信息:
        stream
                .map(new RichMapFunction<TradeCourseOrderBean, TradeCourseOrderBean>() {

                    private Connection conn;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        conn = JdbcUtil.getPhoenixConnection();
                    }

                    @Override
                    public TradeCourseOrderBean map(TradeCourseOrderBean bean) throws Exception {
                        // 根据一个 id 去对应的维度表中查询
                        // 1. provinceInfo
                        JSONObject provinceInfo = DimUtil.readDimFromPhoenix(conn, "dim_base_province", bean.getProvinceId());
                        bean.setProvinceName(provinceInfo.getString("NAME"));
                        //2.
                        JSONObject subjectIdInfo = DimUtil.readDimFromPhoenix(conn, "dim_course_info", bean.getCourseId());
//                        System.out.println("subjectIdInfo"+subjectIdInfo.toJSONString());
                        bean.setSubjectId(subjectIdInfo.getString("SUBJECT_ID"));
                        //...

                        return bean;
                    }
                })
                .print("addDims");
    }

    private SingleOutputStreamOperator<TradeCourseOrderBean> windowAndAgg(
            SingleOutputStreamOperator<TradeCourseOrderBean> beanStream) {
        return beanStream
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<TradeCourseOrderBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner((bean, ts) -> bean.getTs())
                )
                .keyBy(TradeCourseOrderBean::getCourseId)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(
                        new ReduceFunction<TradeCourseOrderBean>() {
                            @Override
                            public TradeCourseOrderBean reduce(TradeCourseOrderBean b1,
                                                               TradeCourseOrderBean b2) throws Exception {
                                b1.setOrderOriginTotalAmount(b1.getOrderOriginTotalAmount().add(b2.getOrderOriginTotalAmount()));
                                b1.setOrderAmount(b1.getOrderAmount().add(b2.getOrderAmount()));
                                b1.setOrderCouponReduceAmount(b1.getOrderCouponReduceAmount().add(b2.getOrderCouponReduceAmount()));
                                return b1;
                            }
                        },
                        new ProcessWindowFunction<TradeCourseOrderBean, TradeCourseOrderBean, String, TimeWindow>() {
                            @Override
                            public void process(String skuId,
                                                Context ctx,
                                                Iterable<TradeCourseOrderBean> elements,
                                                Collector<TradeCourseOrderBean> out) throws Exception {
                                TradeCourseOrderBean bean = elements.iterator().next();
                                bean.setStt(AtguiguUtil.toDateTime(ctx.window().getStart()));
                                bean.setEdt(AtguiguUtil.toDateTime(ctx.window().getEnd()));

                                //                        bean.setTs(System.currentTimeMillis());
                                bean.setTs(ctx.currentProcessingTime());

                                out.collect(bean);

                            }
                        }

                );
    }

    private SingleOutputStreamOperator<TradeCourseOrderBean> parseToPOJO(
            SingleOutputStreamOperator<JSONObject> distinctedStream) {
        return distinctedStream
                .map(obj -> TradeCourseOrderBean.builder()
                        .provinceId(obj.getString("province_id"))
                        .courseId(obj.getString("course_id"))
                        .courseName(obj.getString("course_name"))
                        .orderOriginTotalAmount(obj.getBigDecimal("origin_amount"))
                        .orderAmount(obj.getBigDecimal("final_amount"))
                        .orderCouponReduceAmount(obj.getBigDecimal("coupon_reduce") == null ? new BigDecimal("0") : obj.getBigDecimal("split_coupon_amount"))
                        .ts(obj.getLong("ts") * 1000) // 把s 变成 ms
                        .build());
    }

    private SingleOutputStreamOperator<JSONObject> distinct(DataStreamSource<String> stream) {
        /*
        去重思路:
            目的 要 row_op_ts 最大的那个, 最全的, 也一定是最后来的那个
            
        思路 1:
            没有办法从个数上记录, 找到最后一个
            
            数据是同时产生, 由于网络的波动导致的数据重复
            不管详情有多少条重复, 最终 5s 内一定会来齐
            
            第一个来的的时候, 注册一个定时器:5s 后触发的定时器(处理时间)
                当定时器触发的时候, 同一个详情的所有重复数据来齐.
                找到row_op_ts最大的那个
                提高效率: 每来一个都比较得到最大的那个
                
            
           最终结果: 第一个来了之后, 5s 后处结果
                
        思路 2:
            窗口: 把同一个详情的重复数据, 放入到同一个窗口中, 等到窗口触发计算的时候, 排序找到最大的.
            
            基于时间的窗口: 滚动的基于事件时间, 窗口长度: 1s
                会话窗口: gap 5s
                    最后来了之后,等待 5s 之后出结果
                    
        思路 3:
            如果要汇总的指标只在左表中, 有表的没用.无需等最完整的那条数据
            
            换句话说过, 第一条不管全不全, 都满足需求. 所以只取第一条
                
            
            
         */
        return stream
                .map(JSON::parseObject)
                .keyBy(obj -> obj.getString("id"))
                // 要用定时器: 必须先 keyBy
                .process(new KeyedProcessFunction<String, JSONObject, JSONObject>() {

                    private ValueState<JSONObject> dataState;
                    private ValueState<Boolean> isFirstState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        isFirstState = getRuntimeContext().getState(new ValueStateDescriptor<Boolean>("isFirstState", Boolean.class));
                        dataState = getRuntimeContext().getState(new ValueStateDescriptor<JSONObject>("dataState", JSONObject.class));
                    }

                    @Override
                    public void onTimer(long timestamp,
                                        OnTimerContext ctx,
                                        Collector<JSONObject> out) throws Exception {
                        // 定时器触发的时候执行这个方法.
                        // 当定时器触发的时候, 状态中就是最新的那个
                        // 把状态中的值输出
                        out.collect(dataState.value());
                    }

                    @Override
                    public void processElement(JSONObject value,
                                               Context ctx,
                                               Collector<JSONObject> out) throws Exception {

                        // 当这个详情的第一条数据进来的时候, 注册一个定时器: 5s 后触发的定时器
                        if (isFirstState.value() == null) {
                            // 更新状态
                            isFirstState.update(false);
                            ctx.timerService().registerProcessingTimeTimer(ctx.timerService().currentProcessingTime() + 5000);
                            // 存储数据:
                            dataState.update(value);
                        } else {
                            // 不是第一条: 当前和状态中的进行比较, 时间大的存入到状态中
                        /*
                        2022-10-11 07:42:48.491Z
                        2022-10-11 07:42:48.49Z  // 490
                        2022-10-11 07:42:48.5Z   // 500ms
                        2022-10-11 07:42:48.005Z   // 5ms
                        
                        2022-10-11 07:42:48.51
                        2022-10-11 07:42:48.5
                         */
                            String last = dataState.value().getString("row_op_ts").replaceAll("Z", "");
                            String current = value.getString("row_op_ts").replaceAll("Z", "");
                            // 新数据的时间大于旧数据, 就更新状态
                            if (current.compareTo(last) > 0) {
                                dataState.update(value);
                            }
                        }
                    }
                });
    }
}
/*
交易域SKU粒度下单各窗口汇总表

统计各维度各窗口的原始金额、活动减免金额、优惠券减免金额和订单金额

1. 数据源
    dwd 下单事务事实表
        粒度: sku
        
        order_detail
            内连接
        order_info
            左连接
        详情活动
            左连接
        详情优惠券
            lookup 的内连接
        字典表
2. 数据源有左连接, 需要去重
     
     详情1 sku1  100    null
     null
     详情1 sku1  100     20
     
     去重: 按照详情 id 去重, 保留最全的那个. 时间最大的那个
 
3. 把每条数据封装到 pojo 中

4. 按照 sku 分组开窗聚合
   0-5 sku1  100 200 ...

5. 聚合后, 根据 sku_id 去补充其他的维度信息
    
    根据 sku_id 去查找对应的维度: sku_info
    
    select * from sku_info where id=1
        使用 jdbc 去查找
        
        优化: 旁路缓存和异步
        
6. 写出到 clickhouse 中
 

 */